package com.wang.kafkastudy.controller;

import com.wang.kafkastudy.entity.Message;
import com.wang.kafkastudy.handler.KafkaSendResultHandler;
import org.springframework.beans.factory.annotation.Autowired;


import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @program: KafkaStudy
 * @ClassName TestController
 * @description:
 * @author: wang.yang7
 * @date: 2024-07-25 12:34
 * @Version 1.0
 **/

@RestController
@Transactional(rollbackFor = RuntimeException.class)
public class TestController {

    private final KafkaTemplate<Object, Object> kafkaTemplate;


    public TestController(KafkaTemplate<Object, Object> kafkaTemplate, KafkaSendResultHandler kafkaSendResultHandler) {
        this.kafkaTemplate = kafkaTemplate;
        this.kafkaTemplate.setProducerListener(kafkaSendResultHandler);
    }



    @GetMapping("/kafka/normal/{message}")
    public void sendNormalMessage(@PathVariable("message") String message) {
        kafkaTemplate.send("wangyang", message);
    }
    @GetMapping("/send")
    public void send(@RequestBody Message message) {
        kafkaTemplate.send("wangyang", message);

    }
}
